import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.*;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;

public class LogOP {
    HbaseOP hbaseOP;

    public static void main(String[] args) throws Exception {
        LogOP op = new LogOP();
        System.out.println("start op...");
//        OP1(op);
//        op2(op);
        OP3(op);
        op.hbaseOP.close();
        System.out.println("zbk 183080106");
    }

    private static void OP3(LogOP op) throws Exception {
        op.hbaseOP.deleteTable("logOP");
        System.out.println("delete table");
        op.hbaseOP.createTable("logOP", new String[]{"f"});
        System.out.println("create table");
        ArrayList<String[]> data = op.getBathesArray("src/main/resources/hadoop-root-datanode-master.log");
        op.putBatch(data);
    }

    private static void OP1(LogOP op) throws Exception {
        op.hbaseOP.deleteTable("logOP");
        op.hbaseOP.createTable("logOP", new String[]{"f"});
        op.insertLogs("logOP", "src/main/resources/hadoop-root-datanode-master.log");
    }

    private static void op2(LogOP op) throws IOException {
        Map<String, List<Cell>> ERROR_RESMap = op.filtrateLogs("logOP", "f", "level", "ERROR", CompareOperator.EQUAL);
        HashMap<String, Integer> ERROR_Map = op.countRes(ERROR_RESMap);
        Map<String, List<Cell>> FATAL_RESMap = op.filtrateLogs("logOP", "f", "level", "FATAL", CompareOperator.EQUAL);
        HashMap<String, Integer> FATAL_MAP = op.countRes(FATAL_RESMap);
        float total = op.hbaseOP.countTableRows("logOP");
        if (ERROR_Map.size() != 0) {
            System.out.printf("ERROR比例 %.2f%%\n", ERROR_Map.get(ERROR_Map.keySet().iterator().next()) * 100 / total);
        } else {
            System.out.printf("ERROR比例 %.2f%%\n", 0.00);
        }
        if (FATAL_MAP.size() != 0) {
            System.out.printf("FATAL比例 %.2f%%\n", FATAL_MAP.get(FATAL_MAP.keySet().iterator().next()) * 100 / total);
        } else {
            System.out.printf("FATAL比例 %.2f%%\n", 0.00);
        }
        System.out.println("ERROR最多类： " + ERROR_Map.keySet().iterator().next());
    }

    public LogOP() {
        this.hbaseOP = new HbaseOP("logOP");
    }

    public void insertLogs(String logTable, String logPath) throws Exception {
        BufferedReader reader = new BufferedReader(new FileReader(logPath));

        String str;
        String datetime;
        String[] split;
        long tsp;

        String pattern = "^\\d{4}-\\d{2}-\\d{2}\\s\\d{2}:\\d{2}:\\d{2},\\d{3}.*";
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-ddHH:mm:ss,SSS");

        while ((str = reader.readLine()) != null) {
            split = str.split(" ");
            if (str.matches(pattern) && split.length > 4) {
                datetime = str.substring(0, str.indexOf(",") + 4);
                tsp = Long.MAX_VALUE - sdf.parse(datetime).getTime();
                hbaseOP.insertData(logTable, String.valueOf(tsp), "f", "level", split[2].trim());
                hbaseOP.insertData(logTable, String.valueOf(tsp), "f", "class", split[3].replace(":", ""));
                hbaseOP.insertData(logTable, String.valueOf(tsp), "f", "reason", str.substring(str.indexOf(": ") + 2));
                hbaseOP.insertData(logTable, String.valueOf(tsp), "f", "datetime", datetime);
            }
        }

    }

    public Map<String, List<Cell>> filtrateLogs(String tableName, String family, String colName,
                                                String value, CompareOperator operator) throws IOException {
        Table table;
        Scan scan = hbaseOP.scan;

        SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes(family),
                Bytes.toBytes(colName), operator, Bytes.toBytes(value));

        scan.setFilter(filter);

        if (hbaseOP.table == null) {
            table = hbaseOP.connection.getTable(TableName.valueOf(tableName));
        } else {
            table = hbaseOP.table;
        }
        ResultScanner scanner = table.getScanner(scan);
        Map<String, List<Cell>> map = new HashMap<>();

        for (Result result : scanner) {
//            for (Cell cell : result.rawCells()) {
//                System.out.println(
//                        "行键:" + new String(CellUtil.cloneRow(cell)) + "\t" +
//                                "列族:" + new String(CellUtil.cloneFamily(cell)) + "\t" +
//                                "列名:" + new String(CellUtil.cloneQualifier(cell)) + "\t" +
//                                "值:" + new String(CellUtil.cloneValue(cell)) + "\t" +
//                                "时间戳:" + cell.getTimestamp());
//            }
            map.put(Bytes.toString(result.getRow()), result.listCells());
        }
        return map;
    }

    public HashMap<String, Integer> countRes(Map<String, List<Cell>> resMap) {
        Collection<List<Cell>> values = resMap.values();
        HashMap<String, Integer> map = new HashMap<>();
        values.forEach(list ->
                {
                    Cell cell = list.get(0);
                    String className = new String(CellUtil.cloneValue(cell));
//                    System.out.println(className);
                    if (map.containsKey(className)) {
                        map.replace(className, map.get(className) + 1);
                    } else {
                        map.put(className, 1);
                    }
                }
        );
        if (map.size() != 0) {
            AtomicReference<String> item = new AtomicReference<>(map.keySet().iterator().next());
            final AtomicReference<Integer> value = new AtomicReference<>(map.get(item.get()));
            map.forEach((k, v) -> {
                if (value.get() < v) {
                    item.set(k);
                    value.set(v);
                }
            });
            map.clear();
            map.put(item.get(), value.get());
        }
        return map;
    }

    public void putBatch(ArrayList<String[]> data) throws Exception {
        Table table = hbaseOP.table;
        List<Put> putList = null;
//        table.setAutoFlush(false );
        for (int i = 0; i < data.size(); i++) {
            if (i % 1000 == 0) {
                putList = new ArrayList<>();
            }
            String[] strings = data.get(i);
            Put put = new Put(strings[0].getBytes());
            put.addColumn("f".getBytes(),"level".getBytes(),strings[1].getBytes());
            put.addColumn("f".getBytes(),"class".getBytes(),strings[2].getBytes());
            put.addColumn("f".getBytes(),"reason".getBytes(),strings[3].getBytes());
            put.addColumn("f".getBytes(),"datetime".getBytes(),strings[4].getBytes());
            putList.add(put);
            if (i % 1000 == 999)
            {
                putList.forEach(System.out::println);

                table.put(putList);
//                System.out.println("1000 done");
            }
        }
//        table.flushCommits();
        table.close();
        System.out.println("Batch Data inputted...");
    }

    public HashMap<Long,String[]> getBathesMap(String logPath) throws IOException, ParseException {
        BufferedReader reader = new BufferedReader(new FileReader(logPath));

        String str;
        String datetime;
        String _level;
        String _class;
        StringBuilder _reason;
        String[] split;
        HashMap<Long,String[]> data = new HashMap<>();

        long tsp;

        String pattern = "^\\d{4}-\\d{2}-\\d{2}\\s\\d{2}:\\d{2}:\\d{2},\\d{3}.*";
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-ddHH:mm:ss,SSS");
        str = reader.readLine();
        while (str!= null) {
            split = str.split(" ");
            if (str.matches(pattern) && split.length > 4) {
                datetime = str.substring(0, str.indexOf(",") + 4);
                tsp = Long.MAX_VALUE - sdf.parse(datetime).getTime();
                _level = split[2].trim();
                _class = split[3].replace(":", "");
                _reason = new StringBuilder(str.substring(str.indexOf(": ") + 2));
                if (_reason.toString().contains("IOException in")) {
                   while (!(str = reader.readLine()).matches(pattern)){
                       _reason.append('\n').append(str);
                   }
                   data.put(tsp,new String[]{_level,_class, _reason.toString(),datetime});
                   continue;
                }
                data.put(tsp,new String[]{_level,_class, _reason.toString(),datetime});
            }
            str = reader.readLine();
        }
//        data.forEach((k,v)->{
//            System.out.println(""+k+": "+ Arrays.toString(v));
//        });
//        System.out.println(data.size());
        return data;
    }

    public ArrayList<String[]> getBathesArray(String logPath) throws IOException, ParseException {
        BufferedReader reader = new BufferedReader(new FileReader(logPath));

        String str;
        String datetime;
        String _level;
        String _class;
        StringBuilder _reason;
        String[] split;
        ArrayList<String[]> data = new ArrayList<>();

        long tsp;

        String pattern = "^\\d{4}-\\d{2}-\\d{2}\\s\\d{2}:\\d{2}:\\d{2},\\d{3}.*";
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-ddHH:mm:ss,SSS");
        str = reader.readLine();
        while (str!= null) {
            split = str.split(" ");
            if (str.matches(pattern) && split.length > 4) {
                datetime = str.substring(0, str.indexOf(",") + 4);
                tsp = Long.MAX_VALUE - sdf.parse(datetime).getTime();
                _level = split[2].trim();
                _class = split[3].replace(":", "");
                _reason = new StringBuilder(str.substring(str.indexOf(": ") + 2));
                if (_reason.toString().contains("IOException in")) {
                    while (!(str = reader.readLine()).matches(pattern)){
                        _reason.append('\n').append(str);
                    }
                    data.add(new String[]{String.valueOf(tsp),_level,_class, _reason.toString(),datetime});
                    continue;
                }
                data.add(new String[]{String.valueOf(tsp),_level,_class, _reason.toString(),datetime});
            }
            str = reader.readLine();
        }
        data.forEach(v->System.out.println(Arrays.toString(v)));
//        System.out.println(data.size());
        return data;
    }

}
